package com.zhss.microservice.server.network;

import com.zhss.microservice.server.constant.NodeStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.DataInputStream;
import java.io.IOException;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * Socket读IO线程
 */
public class ServerReadIOThread extends Thread {

    private static final Logger LOGGER = LoggerFactory.getLogger(ServerReadIOThread.class);

    /**
     * master节点间的网络连接
     */
    private Socket socket;
    /**
     * 从远程节点读取数据的输入流
     */
    private DataInputStream inputStream;
    /**
     * 消息接收队列
     */
    private LinkedBlockingQueue<ByteBuffer> receiveQueue;

    /**
     * 构造函数
     * @param socket
     */
    public ServerReadIOThread(
            Socket socket,
            LinkedBlockingQueue<ByteBuffer> receiveQueue) {
        this.socket = socket;
        try {
            this.inputStream = new DataInputStream(socket.getInputStream());
            this.receiveQueue = receiveQueue;
        } catch (IOException e) {
            LOGGER.error("get input stream from socket error......", e);
        }
    }

    /**
     * 线程运行逻辑
     */
    @Override
    public void run() {
        while(NodeStatus.isRunning()) {
            try {
                // 从IO流里读取一条消息
                int messageLength = inputStream.readInt();
                byte[] messageBytes = new byte[messageLength];
                inputStream.readFully(messageBytes, 0, messageLength);

                // 将消息字节数组封装成ByteBuffer
                ByteBuffer message = ByteBuffer.wrap(messageBytes);

                // 将消息放入接收队列里去
                receiveQueue.put(message);

                LOGGER.info("从远程server节点读取到一条消息：" + socket.getRemoteSocketAddress() + "......");
            } catch (IOException e) {
                LOGGER.error("read message from remote node error......", e);
                NodeStatus.fatal();
            } catch (InterruptedException e) {
                LOGGER.error("put message into receive queue error......", e);
                NodeStatus.fatal();
            }
        }

        if(NodeStatus.isFatal()) {
            LOGGER.error("read io thread encounters fatal exception, system is going to shutdown......");
        }
    }

}
